package com.example.springtask.config;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.integration.chunk.ChunkRequest;
import org.springframework.batch.integration.chunk.RemoteChunkingManagerStepBuilderFactory;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
import org.springframework.batch.integration.launch.JobLaunchingGateway;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.PassThroughLineMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.env.Environment;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.kafka.dsl.Kafka;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.MessageHandler;

import java.util.Date;
import java.util.Random;
import java.util.function.Supplier;

@Profile("!worker")
@Configuration
@EnableBatchIntegration
@EnableBatchProcessing
public class ManageConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private RemoteChunkingManagerStepBuilderFactory stepBuilderFactory;

    @Bean
    public Job remoteChunkingJob() throws Exception {
        Random random = new Random();
        return this.jobBuilderFactory.get("remoteChunkingJob" + random.nextInt())
                .start(masterStep())
                .listener(new JobExecutionListener() {
                    @Override
                    public void beforeJob(JobExecution jobExecution) {

                    }

                    @Override
                    public void afterJob(JobExecution jobExecution) {
                        System.out.println("执行结果：" + jobExecution.getStatus());
                    }
                })
                .build();
    }

    public TaskletStep masterStep() {
        return stepBuilderFactory.get("masterStep")
                .chunk(2)
                .reader(sampleReader())
                .inputChannel(replies())
                .outputChannel(requests())
                .build();
    }


    @Bean
    public IntegrationFlow inboundFlow(ConsumerFactory consumerFactory){
        return IntegrationFlows.from(
                Kafka.messageDrivenChannelAdapter(consumerFactory, "replies")
        ).channel(replies()).get();
    }

    @Bean
    public IntegrationFlow outboundFlow(KafkaTemplate kafkaTemplate){
        return IntegrationFlows.from(requests())
                .handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic("requests"))
                .get();
    }

    @Bean
    public QueueChannel replies() {
        return new QueueChannel();
    }

    @Bean
    public DirectChannel requests() {
        return new DirectChannel();
    }

    @Bean
    @StepScope
    public FlatFileItemReader<String> sampleReader() {
        Resource resource = new ClassPathResource("sample.txt");
        FlatFileItemReader<String> flatFileItemReader = new FlatFileItemReader<>();
        flatFileItemReader.setEncoding("utf-8");
        flatFileItemReader.setResource(resource);
        flatFileItemReader.setLineMapper(new PassThroughLineMapper());
        return flatFileItemReader;
    }

}
